package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.date.BetweenFormater;
import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.sparing.async.AsyncTask;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.CreateTableDTO;
import com.hexinfo.dmpro.sparing.dto.DistCPDTO;
import com.hexinfo.dmpro.sparing.dto.IgnoreListDTO;
import com.hexinfo.dmpro.sparing.model.HdfsHeatBackup;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibrary;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibraryTask;
import com.hexinfo.dmpro.sparing.service.*;
import com.hexinfo.dmpro.sparing.util.Logger;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.Future;

@Service
@AllArgsConstructor
@Slf4j
public class LibraryLevelRetryServiceImpl implements LibraryLevelRetryService{

    private final ScanMetadataService scanMetadataService;

    private final SyncCompleteLibraryService syncCompleteLibraryService;

    private final SyncCompleteLibraryTaskService syncCompleteLibraryTaskService;

    private final ImpalaService impalaService;

    private final DistCPService distCPService;

    private final DataRefreshService dataRefreshService;

    //稽核所需SQL模板
    private final String auditSql = "select count(1) as count from {} ";
    @Override
    public String levelRetry(List<SyncCompleteLibrary> list) {
        String result = "success";
        for (SyncCompleteLibrary library : list){
            //表名包含【 为同步任务和自动建表任务 不触发重试功能
            if(library.getLibraryTableName().contains("【")){
                return result;
            }
            //实时更新任务信息
            SyncCompleteLibraryTask task = syncCompleteLibraryTaskService.getBaseMapper().selectById(library.getTaskId());
            //更新任务信息
            if(!ScanCommonConstants.ClusterName.CPZ.value.equals(task.getTaskState())){
                task.setTaskState(ScanCommonConstants.ClusterName.CPZ.value);
                syncCompleteLibraryTaskService.saveOrUpdate(task);
            }
            //当前开始时间
            Date staDate = DateUtil.date();
            List<String> split = StrUtil.split(library.getLibraryTableName(),".",-1,true,true);
            if(split.size() == 2){
                library.setTableName(split.get(1));
                library.setDbName(split.get(0));
            }
            Long sourceCountNumber = library.getSourceCountNumber();
            Long targetCountNumber = library.getTargetCountNumber();
            if("aresoft".equals(library.getDbName())){
                // aresoft库 重试是只做稽核
                library.setTaskState(ScanCommonConstants.ClusterName.CPZ.value);
                syncCompleteLibraryService.saveOrUpdate(library);
                levelRetryAudit(library);
            }else{
                //建表
                levelRetryCreateTable(library);
                //同步、刷新、稽核
                levelRetryDistcp(library);
            }
            //更新用时
            library.setHotStandbyTime(DateUtil.formatBetween(staDate,DateUtil.date(), BetweenFormater.Level.SECOND));
            syncCompleteLibraryService.saveOrUpdate(library);
            if(library.getTaskState().equals(ScanCommonConstants.ClusterName.SB.value)){
                result += "/n库表："+library.getLibraryTableName()+" 重试失败";
                task.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            }
            //数据总量
            task.setSourceCountNumber(task.getSourceCountNumber()-sourceCountNumber+library.getSourceCountNumber());
            task.setTargetCountNumber(task.getTargetCountNumber()-targetCountNumber+library.getTargetCountNumber());
            //更新任务状态
            //判断是否存在进行中(重跑中)和失败任务
            LambdaQueryWrapper<SyncCompleteLibrary> wrapper = Wrappers.lambdaQuery();
            wrapper.eq(StrUtil.isNotBlank(library.getTaskId()),SyncCompleteLibrary::getTaskId,library.getTaskId())
                    //排除自动建表任务和同步任务
                    .notLike(SyncCompleteLibrary::getLibraryTableName,"【")
                    .apply(StrUtil.isNotBlank(library.getTaskState()),"TASK_STATE in ('"+ ScanCommonConstants.ClusterName.JXZ.value +"','"+ScanCommonConstants.ClusterName.CPZ.value + "','" + ScanCommonConstants.ClusterName.SB.value+"')")
                    .eq(SyncCompleteLibrary::getDel,CommonConstants.STATUS_NORMAL);
            //不存在进行中和失败任务
            if(syncCompleteLibraryService.getBaseMapper().selectCount(wrapper)==0){
                task.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
            }
            syncCompleteLibraryTaskService.saveOrUpdate(task);
        }
        return result;
    }
    /**
     * @Method levelRetryCreateTable
     * @Param library
     * @Return void
     * @Description 重试建表
     * @Author Wei MinYe
     * @Date 2024/6/12 14:55
     * @Version V1.0
     */
    @Override
    public String levelRetryCreateTable(SyncCompleteLibrary library) {
        try {
            library.setLog(library.getLog()+"\n["+ DateUtil.now()+"]【自动化建表开始】");
            library.setLog(library.getLog()+"\n"+library.getLibraryTableName());
            library.setTaskState(ScanCommonConstants.ClusterName.CPZ.value);
            syncCompleteLibraryService.saveOrUpdate(library);
            Map<String,Object> mapJQ = impalaService.showTableHive(library.getLibraryTableName());
            String showJQ = (String)mapJQ.get("error");
            String createSQLWG = "";
            if(StrUtil.isNotBlank(showJQ)){
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【查询建表SQL异常】"+showJQ);
            }else{
                String createWG = (String)mapJQ.get("show");
                createSQLWG = createWG.replace("CREATE TABLE","CREATE TABLE IF NOT EXISTS")
                        .replace("CREATE VIEW","CREATE VIEW IF NOT EXISTS")
                        .replace("CREATE EXTERNAL TABLE","CREATE EXTERNAL TABLE IF NOT EXISTS");
                Map<String,Object> mapWG = impalaService.createTableHive(createSQLWG,library.getLibraryTableName());
                String sql = (String)mapWG.get("sql");
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【建表SQL】\n"+sql);
                String errprWG = (String)mapWG.get("error");
                if(StrUtil.isNotBlank(errprWG)){
                    library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【建表SQL异常】"+library.getLibraryTableName()+"\n"+errprWG);
                }else{
                    library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【建表成功】"+library.getLibraryTableName());
                }
            }
            library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【自动化建表结束】");
            Logger.log(library.getLog());
            syncCompleteLibraryService.saveOrUpdate(library);
        } catch (Exception e) {
            log.error("自动化建表异常",e);
        }
        return "success";
    }
    /**
     * @Method levelRetryDistcp
     * @Param library
     * @Return void
     * @Description 重试Distcp、刷新
     * @Author Wei MinYe
     * @Date 2024/6/12 15:16
     * @Version V1.0
     */
    @Override
    public String levelRetryDistcp(SyncCompleteLibrary library) {
        try {
            //查询集群信息
            LambdaQueryWrapper<ClusterSparingConf> jQWrapper = Wrappers.lambdaQuery();
            LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
            ClusterSparingConf jQCluster = scanMetadataService.wrapperClusterSparingConf(jQWrapper,ScanCommonConstants.ClusterName.JQ.value);
            ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,ScanCommonConstants.ClusterName.WG.value);
            //HDFS路径拼接
            String  lo = jQCluster.getHdfsPrefixPath() + library.getDbName() + ".db/" + library.getTableName() + "/" ;
            HdfsHeatBackup kup =new HdfsHeatBackup();
            DistCPDTO dist = new DistCPDTO();
            kup.setSourceAddress(lo);
            kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo, jQCluster.getHdfsPrefixPath(), wGCluster.getHdfsPrefixPath()), "/" + library.getDbName() + ".db/", "/" + library.getDbName() + ".db/"));
            kup.setSourceDbName(library.getDbName());
            kup.setTargetDbName(library.getDbName());
            dist.setHdfsHeatBackup(kup);
            library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【同步开始】"+" \n["+DateUtil.now()+"]【源路径】"+kup.getSourceAddress()+" \n["+DateUtil.now()+"]【目标路径】"+ kup.getTargetAddress());
            syncCompleteLibraryService.saveOrUpdate(library);
            //库级同步(重试时到表级路径)
            String err = distCPService.execShellDatabase(dist,library);
            if(err.contains("成功")){
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【同步成功】");
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【表刷新开始】");
                Logger.log(library.getLog());
                //表刷新
                String oa = dataRefreshService.souTableIsFinishDatabase("",library.getLibraryTableName());
                if(oa.contains("异常")){
                    library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【表刷新失败】"+oa);
                    library.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                    syncCompleteLibraryService.saveOrUpdate(library);
                }else{
                    library.setLog(library.getLog()+"\n"+oa);
                    syncCompleteLibraryService.saveOrUpdate(library);
                    levelRetryAudit(library);
                    Logger.log(library.getLog());
                }
            }else{
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【同步失败】"+err);
                library.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            }
        } catch (Exception e) {
            log.error("重试Distcp、刷新异常",e);
        }
        return "success";
    }
    /**
     * @Method levelRetryAudit
     * @Param library
     * @Return void
     * @Description 重试稽核
     * @Author Wei MinYe
     * @Date 2024/6/12 17:05
     * @Version V1.0
     */
    @Override
    public String levelRetryAudit(SyncCompleteLibrary library) {
        try {
            String auditSqlJQ = StrUtil.format(auditSql,library.getLibraryTableName());
            String auditSqlWG = StrUtil.format(auditSql,library.getLibraryTableName());
            library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【表稽核开始】");
            syncCompleteLibraryService.saveOrUpdate(library);
            Map<String,Object> mapJQ = new HashMap<>();
            Map<String,Object> mapWG = new HashMap<>();
            library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【Impala】"+"\n["+DateUtil.now()+"]【金桥稽核SQL】"+auditSqlJQ+"\n["+DateUtil.now()+"]【万国稽核SQL】"+auditSqlWG);
            mapJQ = impalaService.countJQ(auditSqlJQ);
            mapWG = impalaService.count(auditSqlWG);
            String auditMessageJQ = (String)mapJQ.get("error");
            String auditMessageWG = (String)mapWG.get("error");
            if(StrUtil.isNotBlank(auditMessageJQ)||StrUtil.isNotBlank(auditMessageWG)){
                if(StrUtil.isBlank(auditMessageJQ)){
                    Long countJQ  = Long.valueOf(String.valueOf(mapJQ.get("count")));
                    auditMessageJQ = "金桥数据量为【"+countJQ+"】";
                    library.setSourceCountNumber(countJQ);
                }
                if(StrUtil.isBlank(auditMessageWG)){
                    Long countWG  = Long.valueOf(String.valueOf(mapWG.get("count")));
                    auditMessageWG = "万国数据量为【"+countWG+"】";
                    library.setSourceCountNumber(countWG);
                }
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]"+auditMessageJQ+"\n["+DateUtil.now()+"]"+auditMessageWG);
                library.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                syncCompleteLibraryService.saveOrUpdate(library);
            }else{
                Long countJQ  = Long.valueOf(String.valueOf(mapJQ.get("count")));
                Long countWG  = Long.valueOf(String.valueOf(mapWG.get("count")));
                library.setLog(library.getLog()+"\n["+DateUtil.now()+"]"+"金桥数据量为【"+countJQ+"】\n["+DateUtil.now()+"]万国数据量为【"+countWG+"】");
                library.setSourceCountNumber(countJQ);
                library.setTargetCountNumber(countWG);
                if(countJQ.equals(countWG)){
                    library.setTaskState(ScanCommonConstants.ClusterName.CPCG.value);
                    library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【对比成功】\n["+DateUtil.now()+"]【表稽核结束】");
                    syncCompleteLibraryService.saveOrUpdate(library);
                }else{
                    library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【对比失败】\n["+DateUtil.now()+"]【表稽核结束】");
                    library.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                    syncCompleteLibraryService.saveOrUpdate(library);
                }
            }
        } catch (NumberFormatException e) {
            log.error("重试稽核异常",e);
        }
        return "success";
    }
}
